package com.frinder.official.akka.persistence;

import akka.actor.AbstractActor;
import akka.actor.ActorSelection;
import akka.persistence.AbstractPersistentActorWithAtLeastOnceDelivery;

import java.io.Serializable;

/**
 * Created by frinder on 2017/10/11.
 */
public class MyPersistentActor3 extends AbstractPersistentActorWithAtLeastOnceDelivery {
    private final ActorSelection destination;

    public MyPersistentActor3(ActorSelection destination) {
        this.destination = destination;
    }

    @Override
    public String persistenceId() {
        return "persistence-id";
    }

    @Override
    public AbstractActor.Receive createReceive() {
        return receiveBuilder().
                match(String.class, s -> {
                    persist(new MsgSent(s), evt -> updateState(evt));
                }).
                match(Confirm.class, confirm -> {
                    persist(new MsgConfirmed(confirm.deliveryId), evt -> updateState(evt));
                }).
                build();
    }

    @Override
    public Receive createReceiveRecover() {
        return receiveBuilder().
                match(Object.class, evt -> updateState(evt)).build();
    }

    void updateState(Object event) {
        if (event instanceof MsgSent) {
            final MsgSent evt = (MsgSent) event;
            deliver(destination, deliveryId -> new Msg(deliveryId, evt.s));
        } else if (event instanceof MsgConfirmed) {
            final MsgConfirmed evt = (MsgConfirmed) event;
            confirmDelivery(evt.deliveryId);
        }
    }


    public static class Msg implements Serializable {
        private static final long serialVersionUID = 1L;
        public final long deliveryId;
        public final String s;

        public Msg(long deliveryId, String s) {
            this.deliveryId = deliveryId;
            this.s = s;
        }
    }

    public static class Confirm implements Serializable {
        private static final long serialVersionUID = 1L;
        public final long deliveryId;

        public Confirm(long deliveryId) {
            this.deliveryId = deliveryId;
        }
    }


    class MsgSent implements Serializable {
        private static final long serialVersionUID = 1L;
        public final String s;

        public MsgSent(String s) {
            this.s = s;
        }
    }

    class MsgConfirmed implements Serializable {
        private static final long serialVersionUID = 1L;
        public final long deliveryId;

        public MsgConfirmed(long deliveryId) {
            this.deliveryId = deliveryId;
        }
    }
}
